package yrm

import (
	"context"
	"errors"
	"fmt"
)

type Publisher[M any] interface {
	// 订阅发布者发布的消息
	// 当有多个SubsOptions时，将忽略第一个以外的
	Subscribe(...SubsOptions) (Flow[M], error)
}

type SubsOptions struct {
	//订阅请求时的上下文，用于设置获取Flow应答的超时时间
	Ctx context.Context
	//订阅时用的数据，通常为空
	Request any
}

type SubRequest[M any] struct {
	SubsOptions
	replyChan chan SubReply[M]
}

type SubReply[M any] struct {
	flow Flow[M]
	err  error
}

type SimplePublisher[M any] struct {
	onSub chan<- SubRequest[M]
}

type CancelRequest struct {
	FlowID string
}

type DownflowOptions struct {
	// 表示下游流慢消费时，Publisher是否需要丢弃消息
	// 当值为true时： 发布者生产消息的速度大于下游消费速度时会丢弃消息
	// 当值为false时：发布者生产消息的速度不会大于下游消费速度
	DropMsgIfSlowly bool
	//流头数据
	Header any
	//消息通道的大小
	MsgChanSize int
}

type Ctx[M any] struct {
	onSub       <-chan SubRequest[M]
	onCancel    chan CancelRequest
	downflowMap map[string]*SimpleFlow[M]
}

func NewPublisher[M any](handler func(*Ctx[M]) error) Publisher[M] {
	subChan := make(chan SubRequest[M], 1)
	go func() {
		c := &Ctx[M]{
			subChan,
			make(chan CancelRequest, 1),
			map[string]*SimpleFlow[M]{},
		}
		err := handler(c)
		c.close(err)
	}()
	return &SimplePublisher[M]{
		subChan,
	}
}

func DefaultSubOptions() SubsOptions {
	return SubsOptions{}
}

func (p *SimplePublisher[M]) Subscribe(options ...SubsOptions) (Flow[M], error) {
	var opt SubsOptions
	if len(options) == 0 {
		opt = DefaultSubOptions()
	} else {
		opt = options[0]
	}

	ctx := opt.Ctx
	if ctx == nil {
		ctx = context.Background()
		ctx.Err()
	}
	replyChan := make(chan SubReply[M], 1)
	p.onSub <- SubRequest[M]{
		opt,
		replyChan,
	}
	select {
	case <-ctx.Done():
		return nil, ctx.Err()
	case reply := <-replyChan:
		return reply.flow, reply.err
	}
}

func DefaultDownflowOptions() DownflowOptions {
	return DownflowOptions{
		DropMsgIfSlowly: false,
	}
}

func (c *Ctx[M]) CreateDownflow(options ...DownflowOptions) Flow[M] {
	var opt DownflowOptions
	if len(options) == 0 {
		opt = DefaultDownflowOptions()
	} else {
		opt = options[0]
	}
	if opt.MsgChanSize == 0 {
		opt.MsgChanSize = 16
	}
	errChan := make(chan error, 1)
	errChan <- nil
	flowID := fmt.Sprintf("%p", errChan)
	cancel := func() {
		c.onCancel <- CancelRequest{
			FlowID: flowID,
		}
	}
	flow := &SimpleFlow[M]{
		opt,
		make(chan M, opt.MsgChanSize),
		errChan,
		cancel,
	}
	c.downflowMap[flowID] = flow
	return flow
}

func (c *Ctx[M]) OnSub() <-chan SubRequest[M] {
	return c.onSub
}

func (c *Ctx[M]) OnCancel() <-chan CancelRequest {
	return c.onCancel
}

var ErrDownflowCancel = errors.New("ErrDownflowCancel")

func (c *Ctx[M]) Cancel(req CancelRequest) {
	flowID := req.FlowID
	flow := c.downflowMap[flowID]
	if flow == nil {
		return
	}
	flow.End(ErrDownflowCancel)
	delete(c.downflowMap, flowID)
}

func (c *Ctx[M]) Broadcast(msg M) {
	for _, flow := range c.downflowMap {
		flow.Send(msg)
	}
}

func (c *Ctx[M]) close(err error) {
	for _, flow := range c.downflowMap {
		flow.End(err)
	}
}

func (req *SubRequest[M]) Reply(flow Flow[M], err error) {
	req.replyChan <- SubReply[M]{
		flow,
		err,
	}
}
